#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include "logger.hpp"

namespace im
{
class MQClient
{
public:
    using MessageCallback = std::function<void(const char*, size_t)>;
    using ptr = std::shared_ptr<MQClient>;
// 成员变量如果没有默认构造，必须在初始化列表初始化，或者使用智能指针即可在构造函数体内赋值，因为智能指针有默认构造
private:
    struct ev_loop* _loop;
    std::unique_ptr<AMQP::LibEvHandler> _handler;
    std::unique_ptr<AMQP::TcpConnection> _connection;
    std::unique_ptr<AMQP::TcpChannel> _channel;
    std::thread _loop_thread;
    // 当对象生命周期结束后，调用er_break异步通知调用ev_run的线程关闭事件循环
    struct ev_async _async_watcher; 
public:
    MQClient(const std::string& user, const std::string& passwd, const std::string& host)
    {
        // 1. 实例化底层网络通信框架的IO事件监控句柄
        _loop = EV_DEFAULT;
        // 2.1 实例化libEvHandler句柄，将AMQP框架与事件监控关联起来
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        // 2.2 实例化连接对象
        std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
        AMQP::Address address(url);
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
        // 3. 实例化信道对象
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());

        // 4. 创建新线程启动底层网络通信框架
        _loop_thread = std::thread([this](){
            ev_run(_loop, 0);
        });
    }
    ~MQClient()
    {
        ev_async_init(&_async_watcher, watcher_callback);
        ev_async_start(_loop, &_async_watcher);
        ev_async_send(_loop, &_async_watcher);
        _loop_thread.join(); // 关闭事件循环之后就等待线程退出
        _loop = nullptr;     // loop置为空
    }
    // 用来异步关闭事件循环的回调函数
    static void watcher_callback(struct ev_loop* loop, ev_async* watcher, int32_t revents)
    {
        ev_break(loop, EVBREAK_ALL);
    }
    // 声明交换机、队列、绑定交换机与队列，共三个工作
    void declareComponents(const std::string& exchange_name, const std::string& queue_name,
                         const std::string& binding_key = "routing_key", AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct)
    {
        // 1. 声明交换机(没有就创建，有就返回)，由于Deferred类删除了拷贝构造函数，所以我们直接用返回值连续操作
        _channel->declareExchange(exchange_name, echange_type)
        .onSuccess([exchange_name](){
            LOG_INFO("{} 交换机创建成功", exchange_name);
        })
        .onError([exchange_name](const char* message){
            LOG_ERROR("声明{}交换机失败: {}", exchange_name, message);
            exit(0);
        });
        // 2. 声明队列(没有就创建，有就返回)
        _channel->declareQueue(queue_name)
            .onError([queue_name](const char* message) {
                LOG_ERROR("声明{}队列失败: {}", queue_name, message);
                exit(0);
            })
            .onSuccess([queue_name](){
                LOG_INFO("{} 队列创建成功", queue_name);
            });
        // 3. 将交换机与队列绑定
        _channel->bindQueue(exchange_name, queue_name, binding_key)
            .onError([exchange_name, queue_name](const char *message) {
                LOG_ERROR("{} - {} 绑定失败：{}", exchange_name, queue_name, message);
                exit(0);
            })
            .onSuccess([exchange_name, queue_name, binding_key](){
                LOG_INFO("{} - {} - {} 绑定成功", exchange_name, queue_name, binding_key);
            });
    }
    // 发布消息
    bool publish(const std::string& exchange_name, const std::string& msg, const std::string& routing_key = "routing_key") 
    {
        LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange_name, routing_key);
        bool ret = _channel->publish(exchange_name, routing_key, msg);
        if (ret == false) {
            LOG_ERROR("{} 发布消息失败：", exchange_name);
            return false;
        }
        return true;
    }
    // 订阅消息队列
    void consume(const std::string& queue_name, const MessageCallback& cb) 
    {
        LOG_DEBUG("开始订阅 {} 队列消息！", queue_name);
        _channel->consume(queue_name, "consume-tag")  //返回值 DeferredConsumer
            .onReceived([this, cb](const AMQP::Message& message, uint64_t deliveryTag, bool redelivered) {
                // 1. 用户传递的cb回调函数，用来业务处理
                cb(message.body(), message.bodySize());
                // 2. 根据成员变量channel调用ack函数
                _channel->ack(deliveryTag);
            })
            .onError([queue_name](const char* message){
                LOG_ERROR("订阅 {} 队列消息失败: {}", queue_name, message);
                exit(0);
            });
    }
};
}